#include <QCoreApplication>
#include <QFile>
#include <QTextStream>
#include <QTimer>
#include <QUrlQuery>
#include <QDebug>
#include <QJsonDocument>
#include <QDateTime>
#include <QJsonArray>

#include "qtc_alarm_core/AlarmEvent.h"
#include "qtc_alarm_core/JsonUtil.h"
#include "qtc_alarm_core/AlarmTree.h"
#include "qtc_alarm_core/AlarmTable.h"
#include "qtc_alarm_persist/ESSink.h"
#include "qtc_alarm_rest/AlarmHistoryClient.h"
#include "qtc_alarm_kafka/AlarmKafkaConsumer.h"

static QFile g_log_file;
static QFile g_events_jsonl_file;
static QFile g_tree_json_file;
static QFile g_table_json_file;

void messageHandler(QtMsgType type, const QMessageLogContext&, const QString &msg) {
  if (!g_log_file.isOpen()) return;
  QTextStream ts(&g_log_file);
  ts << msg << '\n';
  ts.flush();
}

int main(int argc, char** argv) {
  QCoreApplication app(argc, argv);

  // Setup log file
  g_log_file.setFileName("alarm_demo.log");
  g_log_file.open(QIODevice::Append | QIODevice::Text);
  qInstallMessageHandler(messageHandler);
  qInfo() << "alarm_demo starting";

  // Prepare JSON files for detailed outputs
  g_events_jsonl_file.setFileName("alarm_events.jsonl");
  g_events_jsonl_file.open(QIODevice::Append | QIODevice::Text);
  g_tree_json_file.setFileName("alarm_tree.json");
  g_table_json_file.setFileName("alarm_table.json");

  // Prepare ESSink in dry-run mode (write bulk NDJSON to essink_bulk.log)
  qtc_alarm_persist::ESSink::Config sink_cfg;
  sink_cfg.es_uri = "http://localhost:9200";
  sink_cfg.index_prefix = "Accelerator";
  sink_cfg.bulk_size = 3;
  sink_cfg.flush_interval_ms = 1000;
  sink_cfg.dry_run = true; // only log payload, do not POST
  sink_cfg.log_file = "essink_bulk.log";
  qtc_alarm_persist::ESSink sink(sink_cfg);
  sink.start();
  QObject::connect(&sink, &qtc_alarm_persist::ESSink::info, [](const QString& s){ qInfo() << s; });
  QObject::connect(&sink, &qtc_alarm_persist::ESSink::error, [](const QString& s){ qWarning() << s; });

  // Real-time ingestion: Kafka consumer -> AlarmTree & AlarmTable
  qtc_alarm_core::AlarmTree tree("Accelerator");
  qtc_alarm_core::AlarmTable table;
  qtc_alarm_kafka::AlarmKafkaConsumer::Config kcfg;
  kcfg.broker = "192.168.0.102:9092"; // 使用后端正在运行的Kafka
  kcfg.topic_config = "Accelerator";
  // 使用唯一的 group.id，确保每次运行都能从最早的偏移读取
  kcfg.group_id = QString("QtClient-Demo-%1").arg(QDateTime::currentDateTime().toString("yyyyMMddhhmmss"));
  kcfg.root_filter = "Accelerator";
  qtc_alarm_kafka::AlarmKafkaConsumer consumer(kcfg, [&](const qtc_alarm_core::AlarmEvent& ev){
    if (ev.type == "state") {
      // Update tree
      tree.applyEvent(ev);
      // Also update table from raw JSON via ESSink formatting
      // Reconstruct JSON from AlarmEvent minimal fields
      QJsonObject obj;
      obj["severity"] = ev.severity;
      obj["message"] = ev.message;
      obj["time_epoch_ms"] = static_cast<double>(ev.time_epoch_ms);
      table.applyStateJson(ev.pv, obj);
    }
    // Optionally append to sink in dry-run for ES payload visibility
    sink.append(ev);

    // Write per-event JSON line for detailed inspection
    if (g_events_jsonl_file.isOpen()) {
      QJsonObject evobj;
      evobj["type"] = ev.type;
      evobj["root"] = ev.root;
      evobj["config"] = ev.config;
      evobj["pv"] = ev.pv; // colon-separated format
      evobj["severity"] = ev.severity;
      evobj["message"] = ev.message;
      evobj["time_epoch_ms"] = static_cast<double>(ev.time_epoch_ms);
      QJsonDocument evdoc(evobj);
      QTextStream ets(&g_events_jsonl_file);
      ets << QString::fromUtf8(evdoc.toJson(QJsonDocument::Compact)) << '\n';
      ets.flush();
    }
  });
  consumer.start();

  // Periodically dump snapshots similar to Phoebus views
  QTimer snapshot_timer;
  QObject::connect(&snapshot_timer, &QTimer::timeout, [&](){
    // Build raw docs
    QJsonDocument tree_doc(tree.toJson());
    QJsonDocument table_doc(table.toJsonArray());

    // Create pv->row map from table for annotation
    QMap<QString, QJsonObject> rows_by_pv;
    const QJsonArray table_arr = table_doc.array();
    for (const auto& v : table_arr) {
      if (!v.isObject()) continue;
      const auto row = v.toObject();
      const QString pv = row.value("pv").toString();
      if (!pv.isEmpty()) rows_by_pv[pv] = row;
    }

    // Recursively annotate tree leaves with message/time based on pv path
    std::function<void(QJsonObject&, QStringList)> annotate = [&](QJsonObject& node, QStringList path){
      const QString name = node.value("name").toString();
      if (!name.isEmpty()) path.push_back(name);
      const bool has_children = node.contains("children") && node.value("children").isArray();
      if (has_children) {
        QJsonArray children = node.value("children").toArray();
        for (int i=0; i<children.size(); ++i) {
          if (!children[i].isObject()) continue;
          QJsonObject child = children[i].toObject();
          annotate(child, path);
          children[i] = child;
        }
        node["children"] = children;
      } else {
        // Leaf: build pv from path excluding root (first element)
        if (path.size() >= 2) {
          const QString pv = path.mid(1).join(":");
          if (rows_by_pv.contains(pv)) {
            const QJsonObject row = rows_by_pv[pv];
            node["message"] = row.value("message");
            node["time_epoch_ms"] = row.value("time_epoch_ms");
            node["acknowledged"] = row.value("acknowledged");
            node["value"] = row.value("value");
            // Compose Phoebus-like title for leaf: "PV: <name> - <SEVERITY> <MESSAGE>"
            QString sev = row.value("severity").toString();
            if (sev.isEmpty()) sev = node.value("severity").toString();
            const QString msg = row.value("message").toString();
            const QString title = msg.isEmpty()
                                  ? QString("PV: %1 - %2").arg(name, sev)
                                  : QString("PV: %1 - %2 %3").arg(name, sev, msg);
            node["title"] = title;
          } else {
            // Fallback title when table row not found
            const QString sev = node.value("severity").toString();
            node["title"] = QString("PV: %1 - %2").arg(name, sev);
          }
        }
      }
    };
    QJsonObject annotated_tree = tree_doc.object();
    annotate(annotated_tree, {});
    QJsonDocument annotated_tree_doc(annotated_tree);
    qInfo() << "Alarm Tree Snapshot:";
    qInfo() << QString::fromUtf8(annotated_tree_doc.toJson(QJsonDocument::Indented));
    qInfo() << "Alarm Table Snapshot (rows):";
    qInfo() << QString::fromUtf8(table_doc.toJson(QJsonDocument::Indented));

    // Also write snapshots to JSON files for external inspection
    if (g_tree_json_file.open(QIODevice::WriteOnly | QIODevice::Text)) {
      QTextStream ts(&g_tree_json_file);
      ts << QString::fromUtf8(annotated_tree_doc.toJson(QJsonDocument::Indented));
      ts.flush();
      g_tree_json_file.close();
    }
    if (g_table_json_file.open(QIODevice::WriteOnly | QIODevice::Text)) {
      QTextStream ts(&g_table_json_file);
      ts << QString::fromUtf8(table_doc.toJson(QJsonDocument::Indented));
      ts.flush();
      g_table_json_file.close();
    }
  });
  snapshot_timer.start(3000);

  // REST history query (requires alarm logging service running on 9000)
  qtc_alarm_rest::AlarmHistoryClient::Config rest_cfg;
  rest_cfg.service_uri = "http://localhost:9000";
  qtc_alarm_rest::AlarmHistoryClient client(rest_cfg);
  QObject::connect(&client, &qtc_alarm_rest::AlarmHistoryClient::resultReceived, [](const QByteArray& data){
    qInfo() << "REST result bytes:" << data.size();
    qInfo() << QString::fromUtf8(data).left(500);
  });
  QObject::connect(&client, &qtc_alarm_rest::AlarmHistoryClient::error, [](const QString& err){
    qWarning() << "REST error:" << err;
  });

  QUrlQuery params;
  params.addQueryItem("root", "DemoRoot");
  params.addQueryItem("start", "now-1d");
  params.addQueryItem("end", "now");
  params.addQueryItem("size", "10");
  client.search(params);

  QTimer::singleShot(12000, [&](){ consumer.stop(); consumer.wait(); sink.stop(); qInfo() << "alarm_demo done"; app.quit(); });
  return app.exec();
}